package com.sh.data.engine.application.service.datadev;

import cn.hutool.core.convert.Convert;
import com.sh.data.engine.application.model.dto.datadev.FlinkFunctionDetailResponse;
import com.sh.data.engine.application.model.dto.datadev.FlinkResourceListResponse;
import com.sh.data.engine.application.model.request.datadev.SaveFunctionNodeRequest;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkDirTreeDomain;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkNodeFunctionDomain;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkResourceListDomain;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeConfigService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeService;
import com.sh.data.engine.domain.util.ConvertUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @author: mengzheng.mhc
 * @date: 2024/8/9 18:08
 */
@Slf4j
@Service
public class FlinkFunctionApplicationService {

    @Autowired
    private FlinkNodeConfigService flinkNodeConfigService;

    @Autowired
    private FlinkNodeService flinkNodeService;

    public boolean saveFunctionNode(SaveFunctionNodeRequest request) {
        if (null == request.getPid()) {
            // 顶级目录
            request.setPid(3L);
        }
        FlinkNodeFunctionDomain domain = Convert.convert(FlinkNodeFunctionDomain.class, request);
        final Long nodeId = flinkNodeService.saveFunctionNode(domain);
        return true;
    }

    public List<FlinkDirTreeDomain> getDirNodes(Integer type) {
        Long projectId = ServletUtils.getProjectId();
        return flinkNodeService.getDirNodes(type, projectId);
    }

    public FlinkFunctionDetailResponse getFunctionDetail(Long nodeId) {
        FlinkNodeFunctionDomain domain = flinkNodeConfigService.getFunctionDetail(nodeId);
        return Convert.convert(FlinkFunctionDetailResponse.class, domain);
    }

    public List<FlinkResourceListResponse> getAllNodes() {
        final Long projectId = ServletUtils.getProjectId();

        List<FlinkResourceListDomain> resources = flinkNodeService.getResourceByProjectId(projectId);
        return ConvertUtil.copyProperties(resources, FlinkResourceListResponse.class);
    }
}
